import { Connection, ConnectOptions } from './connection';
import { DbProvider } from './db-provider';
import tarn from 'tarn';
import { getProvider } from '../register';
import { SQL } from '../sql-builder';
import { PoolExecutor } from './pool-executor';
import EventEmitter from 'events';

export interface PoolOptions {
  max?: number;
  /**
   * 不能小于零
   */
  min?: number;
  connectTimeout?: number;
  idleTimeout?: number;
}

export const DefaultPoolOptions: PoolOptions = {
  max: 5,
  min: 1,
  connectTimeout: 30000,
  idleTimeout: 30000,
};

export class Pool {
  readonly connectionOptions: ConnectOptions;
  readonly poolOptions: Readonly<Required<PoolOptions>>;
  readonly provider: DbProvider;
  private readonly _emitter = new EventEmitter();

  constructor(
    provider: string | DbProvider,
    poolOptions: PoolOptions,
    connectionOptions: ConnectOptions
  ) {
    if (typeof provider === 'string') {
      this.provider = getProvider(provider);
    } else {
      this.provider = provider;
    }

    this.poolOptions = Object.assign(
      {
        max: 5,
        min: 1,
        connectTimeout: 30000,
        idleTimeout: 30000,
      },
      poolOptions
    );
    this.connectionOptions = connectionOptions;
  }

  on(event: 'healthy-changed', handler: (health: boolean) => void): this;
  on(event: 'acquire', handler: (conn: Connection) => void): this;
  on(event: 'release', handler: (conn: Connection) => void): this;
  on(event: 'connection-open', handler: (conn: Connection) => void): this;
  on(event: 'connection-close', handler: (conn: Connection) => void): this;
  on(event: 'log', handler: (message: string) => void): this;
  on(event: string, handler: (...args: any) => void): this {
    this._emitter.on(event, handler);
    return this;
  }

  off(event: 'healthy-changed', handler: (health: boolean) => void): this;
  off(event: 'acquire', handler: (conn: Connection) => void): this;
  off(event: 'release', handler: (conn: Connection) => void): this;
  off(event: 'connection-open', handler: (conn: Connection) => void): this;
  off(event: 'connection-close', handler: (conn: Connection) => void): this;
  off(event: 'log', handler: (message: string) => void): this;
  off(event: string, handler: (...args: any) => void): this {
    if (!handler) {
      this._emitter.removeAllListeners(event);
    } else {
      this._emitter.off(event, handler);
    }
    return this;
  }

  private pool?: tarn.Pool<Connection>;

  get opened() {
    return !!this.pool;
  }

  private checkOpened() {
    if (!this.opened) {
      throw new Error(`Pool is not opened yet.`);
    }
  }

  private _healthy?: boolean = undefined;
  /**
   * 当前连接池健康状态
   */
  get healthy() {
    this.checkOpened();
    return this._healthy;
  }

  private async test() {
    let conn: Connection;
    try {
      conn = await this._create();
      return true;
    } catch {
      return false;
    } finally {
      if (conn!) {
        await conn.close();
      }
    }
  }

  private async _create() {
    const conn = this.provider.getConnection(this.connectionOptions);
    try {
      await conn.open();
      this.setHealthy(true);
      this._emitter.emit('connection-open', conn);
    } catch (error) {
      if (this.pool!.numUsed() + this.pool!.numFree() <= 0) {
        this.setHealthy(false);
      }
      throw error;
    }
    return conn;
  }

  private setHealthy(healthy: boolean) {
    if (this._healthy !== healthy) {
      this._healthy === healthy;
      this._emitter.emit('healthy-changed', this._healthy);
    }
  }

  async open() {
    await this.test();
    this.pool = new tarn.Pool({
      max: this.poolOptions.max,
      min: this.poolOptions.min,
      createTimeoutMillis: this.poolOptions.connectTimeout,
      log: (msg: string) => {
        this._emitter.emit('log', msg);
      },
      validate: (async (conn: Connection) => {
        try {
          await conn.query(SQL.select(1));
          return true;
        } catch {
          return false;
        }
      }) as any, // tarn decalre file not upgrade
      create: this._create.bind(this),
      destroy: async conn => {
        conn.close().finally(() => {
          this._emitter.emit('connection-close', conn);
        });
      },
      propagateCreateError: true,
    });
  }

  async close() {
    this.checkOpened();
    await this.pool?.destroy();
    this.pool = undefined;
    this._healthy = undefined;
  }

  /**
   * 获取连接
   */
  acquire(): Promise<Connection> {
    this.checkOpened();
    return this.pool!.acquire().promise.then(conn => {
      this._emitter.emit('acquire', conn);
      return conn;
    });
  }

  release(conn: Connection): void {
    this.checkOpened();
    this.pool!.release(conn);
    this._emitter.emit('release', conn);
  }

  private _executor?: PoolExecutor;
  /**
   * 全局执行器，该执行器在执行SQL时会自动请求和释放连接，但不保证所有执行的SQL都在同一个数据库连接中。
   */
  get executor(): PoolExecutor {
    if (!this._executor) {
      this._executor = new PoolExecutor(this);
    }
    return this._executor;
  }

  /**
   * 当前总连接数
   */
  get size() {
    this.checkOpened();
    return (
      this.pool!.numFree() +
      this.pool!.numUsed() +
      this.pool!.numPendingCreates()
    );
  }

  /**
   * 当前可用连接数
   */
  get available() {
    this.checkOpened();
    return this.pool!.numFree();
  }

  /**
   * 当前等待请求数
   */
  get pending() {
    return this.pool!.numPendingAcquires();
  }

  /**
   * 当前使用中连接数
   */
  get borrowed() {
    return this.pool!.numUsed();
  }
}
